package task3.pimpmypipe;

import java.io.PrintStream;
import java.io.StreamCorruptedException;
import java.security.InvalidParameterException;

public abstract class DataEnrichmentFilter<in, out>
  extends AbstractFilter<in, out>
  implements Runnable
{
  private out m_TempWriteEntity = null;
  private boolean m_EndOfStream = false;
  
  public DataEnrichmentFilter(Readable<in> input, Writeable<out> output)
    throws InvalidParameterException
  {
    super(input, output);
  }
  
  public DataEnrichmentFilter(Readable<in> input)
    throws InvalidParameterException
  {
    super(input);
  }
  
  public DataEnrichmentFilter(Writeable<out> output)
    throws InvalidParameterException
  {
    super(output);
  }
  
  public out read()
    throws StreamCorruptedException
  {
    return getNextEntity();
  }
  
  public void write(in value)
    throws StreamCorruptedException
  {
    if (!this.m_EndOfStream) {
      if (value != ENDING_SIGNAL)
      {
        if (this.m_TempWriteEntity == null) {
          this.m_TempWriteEntity = getNewEntityObject();
        }
        if (fillEntity(value, this.m_TempWriteEntity))
        {
          writeOutput(this.m_TempWriteEntity);
          this.m_TempWriteEntity = null;
        }
      }
      else
      {
        if (this.m_TempWriteEntity != null) {
          writeOutput(this.m_TempWriteEntity);
        }
        sendEndSignal();
      }
    }
  }
  
  protected out getNextEntity()
    throws StreamCorruptedException
  {
    if (!this.m_EndOfStream)
    {
      out entity = getNewEntityObject();
      boolean finished = false;
      do
      {
        in input = readInput();
        if (input != null)
        {
          finished = fillEntity(input, entity);
        }
        else
        {
          this.m_EndOfStream = true;
          finished = true;
        }
      } while (!finished);
      return entity;
    }
    return null;
  }
  
  public void run()
  {
    out output = null;
    try
    {
      do
      {
        output = read();
        if (output != null) {
          writeOutput(output);
        }
      } while (output != null);
      sendEndSignal();
    }
    catch (StreamCorruptedException e)
    {
      System.out.print("Thread reports error: ");
      System.out.println(Thread.currentThread().getId() + " (" + Thread.currentThread().getName() + ")");
      e.printStackTrace();
    }
  }
  
  protected abstract boolean fillEntity(in paramin, out paramout);
  
  protected abstract out getNewEntityObject();
}
